feat: Block segment publish action for a streaming task until prior task groups are complete - #20505

[P1] Complete the async response on every path

Finding: doAction starts an AsyncContext, but this helper only sets the status and optionally writes a body; it never completes the context. Consequently, even a finished task action leaves the HTTP exchange open until the async timeout, and the no-leader path returns here before installing any timeout listener.

Suggestion: Complete the async context after the response is written, including when no response body is present.

[P1] Preserve null task-action results

Finding: Several existing actions, including LockReleaseAction and UpdateLocationAction, implement TaskAction<Void> and return null. Map.of rejects null values, so these successful actions throw from onSuccess before completeRequest can send the result, leaving the async request unanswered.

Suggestion: Build the response with a null-tolerant map so the existing result: null response contract is preserved.

[P2] Avoid mutating the map being iterated

Finding: The constructor iterates blockingTaskGroups.keySet() while each completion callback removes from that same constructor argument. If a prior group completes during registration, addCompletionListener can run the callback inline (or concurrently), modifying the HashMap during iteration and throwing ConcurrentModificationException before the readiness future is returned.

Suggestion: Register listeners from an immutable snapshot and update a separate thread-safe tracking structure from the callbacks.

[P2] Keep publish work off the supervisor notice thread

Finding: The readiness future is completed by TaskGroup.onCompleted during checkPendingCompletionTasks, which runs on this supervisor's single-threaded notice executor. With a direct executor, this transform calls perform inline; that method enters the task-lock critical section and performs metadata-store publication, so a slow publish blocks this supervisor from processing subsequent status checks and notices. The same executor choice is present in the insert action.

Suggestion: Dispatch the publish operation to a dedicated executor when readiness completes, leaving the supervisor notice thread free to process its queue.

[P1] Make the async action timeout cover publish waits

Finding: The /action route has no {timeout} path segment, so @PathParam("timeout") is never populated and setTimeout always uses five minutes. The supervisor's default completionTimeout is 30 minutes; a legitimate pending task group that takes longer than five minutes causes the timeout listener to cancel the action and complete the request before the readiness future finishes. The worker then sees an I/O or empty-response failure, and the publish retry path does not retry that timeout, so the new blocking behavior can fail a task instead of waiting for the prior group.

Suggestion: Bind the timeout to a real request parameter or route and propagate a deadline that covers the configured supervisor completion wait, with an explicit retry-safe timeout response.

Typo: this should be query param.

Follow-up assessment

Agreed, @QueryParam("timeout") is the right binding. That fixes the annotation typo, but the timeout concern also needs the caller/default addressed: RemoteTaskActionClient posts to /druid/indexer/v1/action without a timeout query parameter, so changing the annotation alone still leaves worker requests capped at five minutes. A prior task group may legitimately remain pending under the supervisor's default 30-minute completionTimeout; the listener cancels the action instead of returning a retryable SegmentPublishResult.

Suggestion: Have worker requests use a deadline that covers the configured supervisor completion wait (or make the server derive it), and return an explicit response that the publish caller can safely retry when that deadline expires. The existing P1 remains actionable at this head; no additional finding is raised by this follow-up.

Reviewed 7 repository source files for this thread: 3 of the PR's 18 changed files and 4 supporting files. This was a focused static follow-up, not a full PR re-review.

This is an automated review by Codex GPT-5.6-Luna(max)

After addressing the findings or replying to the comments, you can request another review from me to trigger a new automated review.

Removed the parameter since the client is never sending it. A future PR may add a new field inside the task action payload itself. Alternatively, it may even be linked up to the completion timeout.

For now, on timeout, we send back a gateway timeout response so that the client retries (via StandardRetryPolicy).

[P1] Initial publish runs before timeout setup

Finding: submitAsync is invoked before the timeout listener and 15-minute timeout are registered. Local streaming publish actions execute their first metadata publish synchronously inside submitAsync, so a slow first attempt is governed by the container's default timeout and cannot be canceled through this listener.

Suggestion: Install the timeout and one-shot completion state before starting the action, or move the initial publish attempt off the request thread.

updated.

[P1] Publish waits can fail local tasks after five minutes

Finding: Defect: this future can remain pending until prior task groups finish, but local callers invoke it through LocalTaskActionClient.submit, whose async branch waits only five minutes. Trigger: a predecessor remains pending or the publish itself exceeds five minutes; the default supervisor completion timeout is 30 minutes, and SequenceMetadata uses synchronous submit for this action. Impact: the local-runner streaming task receives a wrapped TimeoutException and fails instead of waiting for the predecessor or returning the intended retryable publish result.

Suggestion: Give streaming publish waits a deadline aligned with the supervisor completion window, or keep them out of the generic five-minute async-action path so the task can receive the publish result or retry status.

The default timeout has no been increased to 15 minutes.
Moreover, on timeout, the task will retry. So this failure is okay.
A future patch may tie this up to the completion timeout.

The five-minute issue remains at the current head. SequenceMetadata still calls TaskActionClient.submit, and LocalTaskActionClient.performAction still waits on performAsync(...).get(5, TimeUnit.MINUTES), while the supervisor completion timeout defaults to 30 minutes. A valid predecessor wait can therefore fail the local streaming task after five minutes, and the timeout path does not cancel the pending future, so it can retain the listener and later attempt publication after the task has failed. Please align the action timeout with the supervisor completion window and cancel the future on timeout, or keep this long-lived wait out of the fixed synchronous path.

[P2] Generic async failures lose the error payload

Finding: Defect: the generic failure branch completes a bare 500 response, whereas the previous endpoint returned an error entity. Trigger: any action failure that is not delivered as a direct DruidException, including RuntimeException wrappers from transactional publish actions. Impact: remote task clients and operators receive an empty 500 and lose the actionable message and diagnostics previously exposed.

Suggestion: Preserve the previous JSON error payload and exception logging in the generic failure branch while retaining the DruidException-specific status handling.

Updated.

The response-payload part is fixed, but the generic failure branch still omits the previous log.warn(Throwable, ...) call. The client now receives the error message, yet the Overlord loses the stack trace and server-side diagnostics for generic action failures. Please restore throwable logging before completing the 500 response.

[P1] Later pending generations can deadlock

Finding: Once maxAllowedStops or taskCount allows more than one pending group, pendingCompletionTaskGroups can contain older and newer generations for the same partition. This loop adds every overlapping group except groups containing the task ID; it never filters to predecessors. If an older task is still pushing when a newer group reaches publish, the older publish can wait for the newer group while the newer publish waits for the older one. Neither can complete until the completion timeout removes both, stalling ingestion and causing task restarts.

Suggestion: Associate the task with its pending TaskGroup or generation and register listeners only for groups that precede it; add coverage with two pending generations and a delayed first publish.

Yes, I was wondering about this myself. Let me check if it is possible to check on the generation of the task.

Updated the method in SeekableStreamSupervisor to wait on older task groups only.

[P1] Align the local publish wait timeout

Finding: Streaming transactional actions now return a future that may wait for predecessor completion, but SequenceMetadata still invokes TaskActionClient.submit and LocalTaskActionClient.performAction still waits on performAsync(...).get(5, TimeUnit.MINUTES). A predecessor can legitimately remain pending until the supervisor completion timeout, which defaults to 30 minutes, so a valid task fails after five minutes instead of continuing. The timeout does not cancel the future, leaving its listener active and allowing it to invoke publish after the task has failed.

Suggestion: Give this wait an action-specific deadline tied to the supervisor completion timeout and cancel it on local timeout, or keep the long wait out of the fixed five-minute synchronous path.

SequenceMetadata invokes TaskActionClient.submit on a RemoteTaskActionClient which then submits the actual task action to the Overlord, where OverlordResource.performTaskAction uses LocalTaskActionClient.submitAsync.

[P2] Restore generic action failure logging

Finding: The incremental fix restores the generic 500 error payload, but this branch still does not log the throwable. Generic async action failures therefore lose the stack trace and server-side diagnostics that the previous synchronous handler emitted with log.warn, even though the client receives the message.

Suggestion: Log the throwable in the generic onFailure branch before completing the response.

Added back the log.